package com.hliushi.spark.sql

import org.apache.spark.sql.SparkSession
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/18 22:39
 */
class UntypedTransformation {

  // 1.创建SparkSession
  val spark: SparkSession = SparkSession.builder()
    .appName("type_transformation")
    .master("local[6]")
    .getOrCreate()

  import spark.implicits._


  /**
   *
   */
  @Test
  def select(): Unit = {
    val ds = Seq(Person("zhangsan", 12), Person("lisi", 20), Person("zhangsan", 8)).toDS()


    // select * from ...  先入为主
    // from ... select
    // 在Dataset中, select可以在任何位置调用, 用来选择某些列出现的结果集
    ds.select($"name").show()
    //  +--------+
    //  |    name|
    //  +--------+
    //  |zhangsan|
    //  |    lisi|
    //  |zhangsan|
    //  +--------+

    /**
     * 在SQL语句中, 经常可以在select子句中使用count(age), rand()等函数,
     * 在selectExpr中就可以使用这样的SQL表达式, 同时使用select配合expr函数也可以做到类似的效果
     */
    ds.selectExpr("count(age) as count").show()
    ds.selectExpr("sum(age) as sum").show()
    ds.selectExpr("rand() as random").show()

    //  +--------+
    //  |sum(age)|
    //  +--------+
    //  |      40|
    //  +--------+

    import org.apache.spark.sql.functions._

    //
    ds.select(expr("sum(age)")).show()
  }


  /**
   * withColumn:通过Column对象在Dataset中创建创建一个新的列或者修改原来的列
   * withColumnRenamed:修改列名
   * #
   * def withColumn(colName: String, col: Column): DataFrame
   * #
   * def withColumnRenamed(existingName: String, newName: String): DataFrame
   */
  @Test
  def column(): Unit = {
    val ds = Seq(Person("zhangsan", 12), Person("lisi", 20), Person("zhangsan", 8), Person("", 22)).toDS()

    /**
     * 如果想使用SQL中的函数功能
     * 1.使用functions.xx
     * 2.使用表达式, 可以使用expr("...") 随时随地编写表达式
     */
    import org.apache.spark.sql.functions._

    // select rand() from ...
    ds.withColumn("random", expr("rand()")).show()
    //  +--------+---+-------------------+
    //  |    name|age|             random|
    //  +--------+---+-------------------+
    //  |zhangsan| 12| 0.7061171775859106|
    //  |    lisi| 20| 0.8072621539985018|
    //  |zhangsan|  8|0.20542377673444023|
    //  |        | 22| 0.7643748308808244|
    //  +--------+---+-------------------+


    // 给name列重命名为name_new, 但是还是会新增一个列
    ds.withColumn("name_new", $"name").show()
    //  +--------+---+--------+
    //  |    name|age|name_new|
    //  +--------+---+--------+
    //  |zhangsan| 12|zhangsan|
    //  |    lisi| 20|    lisi|
    //  |zhangsan|  8|zhangsan|
    //  |        | 22|        |
    //  +--------+---+--------+

    //
    ds.withColumn("name_jok", 'name === "").show()
    //  +--------+---+--------+
    //  |    name|age|name_jok|
    //  +--------+---+--------+
    //  |zhangsan| 12|   false|
    //  |    lisi| 20|   false|
    //  |zhangsan|  8|   false|
    //  |        | 22|    true|
    //  +--------+---+--------+

    // 给name列重命名为name_new, 在原来的列进行修改, 不新增列
    ds.withColumnRenamed("name", "name_new").show()
  }


  /**
   *
   */
  @Test
  def groupBy(): Unit = {
    val ds = Seq(Person("zhangsan", 12), Person("lisi", 20), Person("zhangsan", 8), Person("", 22)).toDS()

    /**
     * 为什么groupByKey是有类型的?
     * def groupByKey[K: Encoder](func: T => K): KeyValueGroupedDataset[K, T]
     * .  最主要的原因是因为groupByKey所生成的对象中的算子是有类型的
     * #
     * 为什么groupBy是无类型的?
     * def groupBy(cols: Column*): RelationalGroupedDataset
     * .  因为groupBy所生成的对象的算子是无类型的, 针对列进行处理
     */
    ds.groupByKey((item: Person) => item.name).count()

    import org.apache.spark.sql.functions._
    // 以下三种方式都是一样的, 其中 Dataset类中的avg()和mean()就只是方法名称不同摆了, 全都是求平均值avg
    ds.groupBy($"name").agg(mean("age")).show()
    ds.groupBy($"name").avg("age").show()
    ds.groupBy($"name").mean().show()
    ds.groupBy($"name").sum().show()

    //  +--------+--------+
    //  |    name|avg(age)|
    //  +--------+--------+
    //  |zhangsan|    10.0|
    //  |        |    22.0|
    //  |    lisi|    20.0|
    //  +--------+--------+
    //
    //  +--------+--------+
    //  |    name|sum(age)|
    //  +--------+--------+
    //  |zhangsan|      20|
    //  |        |      22|
    //  |    lisi|      20|
    //  +--------+--------+

  }
}